package com.aoip.telemetry.telemetry.receive.start;

import com.aoip.telemetry.telemetry.core.redis.RedisCache;
import com.aoip.telemetry.telemetry.entity.constant.TelemetryConstants;
import com.aoip.telemetry.telemetry.entity.enums.CacheKeyEnum;
import com.aoip.telemetry.telemetry.entity.protocol.ProtocolBaseEntity;
import com.aoip.telemetry.telemetry.receive.handler.BindStationAddressHandle;
import com.aoip.telemetry.telemetry.receive.handler.HeartbeatHandler;
import com.aoip.telemetry.telemetry.receive.handler.IpFilterRuleHandler;
import com.aoip.telemetry.telemetry.receive.handler.QuitHandler;
import com.aoip.telemetry.telemetry.receive.message.common.AsciiMessageCodecSharable;
import com.aoip.telemetry.telemetry.receive.message.common.HexMessageCodecSharable;
import com.aoip.telemetry.telemetry.receive.message.common.MessageSendMqHandler;
import com.aoip.telemetry.telemetry.receive.protocol.DelimiterBasedDecoder;
import com.aoip.telemetry.telemetry.receive.protocol.FixedLengthDecoder;
import com.aoip.telemetry.telemetry.receive.protocol.LengthFieldDecoder;
import com.aoip.telemetry.telemetry.receive.protocol.LineBasedDecoder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.ipfilter.IpFilterRule;
import io.netty.handler.ipfilter.IpFilterRuleType;
import io.netty.handler.ipfilter.IpSubnetFilterRule;
import io.netty.handler.ipfilter.RuleBasedIpFilter;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * netty基于tcp协议，
 *
 * @author chenbai
 * @version 1.0
 * @date 2021/6/15 9:24
 */
@Service
@Slf4j
public class TcpReceiveService {

    @Resource
    private RedisCache redisCache;

    private NioEventLoopGroup bossEventLoopGroup = new NioEventLoopGroup(2);
    private NioEventLoopGroup workEventLoopGroup = new NioEventLoopGroup(10);

    @Value("${netty.sn-tcp-port}")
    private String snHydrologyConnectTcpPort;

    @Value("${netty.sn-tcp-on-off}")
    private boolean tcpOnOff;

    @PostConstruct
    public void init() {
        Collection<String> keys = redisCache.keys(CacheKeyEnum.PROTOCOL + ":*");
        List<ProtocolBaseEntity> list = new ArrayList<>();
        keys.forEach(item -> {
            ProtocolBaseEntity entity = redisCache.getCacheObject(item);
            if (TelemetryConstants.TCP.equals(entity.getProtocolCommunicate())) {
                list.add(entity);
            }
        });
        if (!tcpOnOff || list.size() < 1) {
            return;
        }

//        List<ProtocolBaseEntity> list = new ArrayList<>();
//        ProtocolBaseEntity entity = new ProtocolBaseEntity();
//        entity.setProtocolCommunicate(TelemetryConstants.TCP);
//        entity.setFrameSeparator("NN");
//        entity.setProtocolCodeFormat(TelemetryConstants.ASCII);
//        entity.setFrameLength(1);
//        entity.setMaxFrameLength(1024);
//        entity.setHasCall(TelemetryConstants.NO);
//        entity.setHasHeartbeat(TelemetryConstants.NO);
//        entity.setMsgSeparationMode("3");
//        entity.setProtocolPorts(snHydrologyConnectTcpPort);
//        list.add(entity);

        List<int[]> portList = new ArrayList<>();
        for (ProtocolBaseEntity item : list) {
            String protocolPort = item.getProtocolPorts();
            if (!StringUtils.hasText(protocolPort)) {
                continue;
            }
            if (protocolPort.indexOf(",") > 0) {
                String[] strings = protocolPort.split(",");
                int[] ports = new int[strings.length];
                for (int i = 0; i < strings.length; i++) {
                    ports[i] = Integer.valueOf(strings[i]);
                }
                portList.add(ports);
                item.setProtocolPort(ports);
            } else {
                int[] ports = new int[]{Integer.valueOf(protocolPort)};
                portList.add(ports);
                item.setProtocolPort(ports);
            }
        }
        if (portList.size() < 1) {
            return;
        }
        ServerBootstrap serverBootstrap = initServerBootstrap(list);
        for (int[] ports : portList) {
            for (Integer port : ports) {
                serverBootstrap.bind(port);
            }
        }
    }

    private ServerBootstrap initServerBootstrap(List<ProtocolBaseEntity> list) {
        HexMessageCodecSharable hexMessageCodec = new HexMessageCodecSharable();
        AsciiMessageCodecSharable asciiMessageCodec = new AsciiMessageCodecSharable();
        QuitHandler quitHandler = new QuitHandler();
        RuleBasedIpFilter ruleBasedIpFilter = new RuleBasedIpFilter(new IpFilterRuleHandler());
        return new ServerBootstrap()
                .group(bossEventLoopGroup, workEventLoopGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        int connectPort = ch.localAddress().getPort();
                        ChannelPipeline pipeline = ch.pipeline();

                        for (ProtocolBaseEntity item : list) {
                            for (Integer port : item.getProtocolPort()) {
                                if (connectPort == port) {
                                    String codeFormat = item.getProtocolCodeFormat();
                                    String msgSeparationMode = item.getMsgSeparationMode();
                                    int frameLength = item.getFrameLength();
                                    int maxFrameLength = item.getMaxFrameLength();
                                    maxFrameLength = maxFrameLength <= 0 ? 1024 : maxFrameLength;
                                    if (StringUtils.hasText(msgSeparationMode)) {
                                        switch (msgSeparationMode) {
                                            case "1":
                                                pipeline.addLast("基于固定长度--粘包半包处理", new FixedLengthDecoder(frameLength));
                                                break;
                                            case "2":
                                                pipeline.addLast("字符串转换", new StringDecoder());
                                                pipeline.addLast("基于回车换行--粘包半包处理", new LineBasedDecoder(maxFrameLength));
                                                break;
                                            case "3":
                                                String frameSeparator = item.getFrameSeparator();
                                                ByteBuf byteBuf = ByteBufAllocator.DEFAULT.heapBuffer();
                                                if (TelemetryConstants.HEX.equals(codeFormat)) {
                                                    byte[] bytes = ByteBufUtil.decodeHexDump(frameSeparator);
                                                    byteBuf.writeBytes(bytes);
                                                } else {
                                                    byte[] bytes = frameSeparator.getBytes();
                                                    byteBuf.writeBytes(bytes);
                                                }
                                                pipeline.addLast("基于特殊字符--粘包半包处理", new DelimiterBasedDecoder(maxFrameLength, false, byteBuf));
                                                break;
                                            case "4":
                                                int lengthFieldOffset = item.getLengthFieldOffset();
                                                int lengthFieldLength = item.getLengthFieldLength();
                                                int lengthAdjustment = item.getLengthAdjustment();
                                                int initialBytesToStrip = item.getInitialBytesToStrip();
                                                int contentAfterLength = item.getContentAfterLength();
                                                pipeline.addLast("粘包半包处理", new LengthFieldDecoder(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, contentAfterLength));
                                                break;
                                            default:
                                        }
                                    }
                                    if (TelemetryConstants.HEX.equals(codeFormat)) {
                                        pipeline.addLast("报文ByteBuf与报文字符串相互转换", hexMessageCodec);
                                    } else {
                                        pipeline.addLast("报文ByteBuf与报文字符串相互转换", asciiMessageCodec);
                                    }
                                    String hasHeartbeat = item.getHasHeartbeat();
                                    if (TelemetryConstants.YES.equals(hasHeartbeat)) {
                                        pipeline.addLast("idleStateHandler", new IdleStateHandler(item.getReadIdle(), item.getWriteIdle(), 0));
                                        pipeline.addLast("心跳检测", new HeartbeatHandler());
                                    }
                                    pipeline.addLast("原始报文数据发送到消息中间件", new MessageSendMqHandler(item.getId()));
                                    String hasCall = item.getHasCall();
                                    if (TelemetryConstants.YES.equals(hasCall)) {
                                        String stCodeIndexRange = item.getStCodeIndexRange();
                                        pipeline.addLast("绑定测站地址用于召测", new BindStationAddressHandle(stCodeIndexRange));
                                        pipeline.addLast(quitHandler);
                                    }
                                    break;
                                }
                            }
                        }
                    }
                });
    }

    @PreDestroy
    public void destroy() throws InterruptedException {
        bossEventLoopGroup.shutdownGracefully().sync();
        workEventLoopGroup.shutdownGracefully().sync();
    }

}

//    //                        pipeline.addLast(new RuleBasedIpFilter(new IpFilterRuleHandler()));
//    String ip = ch.remoteAddress().getHostString();
//    List<String> blackIpList = redisCache.getCacheObject(CacheKeyEnum.BLACK_IP_LIST.getCode());
//                        if (blackIpList.contains(ip)) {
//
//                                }
//                                IpSubnetFilterRule rule1 = new IpSubnetFilterRule("192.168.119.1", 24, IpFilterRuleType.ACCEPT);
//                                IpSubnetFilterRule rule2 = new IpSubnetFilterRule("127.0.0.1", 32, IpFilterRuleType.REJECT);
//                                IpFilterRule rejectAll = new IpFilterRule() {
//@Override
//public boolean matches(InetSocketAddress remoteAddress) {
//        return true;
//        }
//
//@Override
//public IpFilterRuleType ruleType() {
//        return IpFilterRuleType.REJECT;
//        }
//        };
//        RuleBasedIpFilter filter = new RuleBasedIpFilter(rule1, rejectAll );
////                        ch.pipeline().addLast("ipFilter", filter);
//        ch.pipeline().addLast("ipFilter", new RuleBasedIpFilter(new IpFilterRuleHandler()));

